package com.atguigu.cn.dataStream.api

import java.util.concurrent.TimeUnit

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
  * @author: yangShen
  * @Description: Flink 的ProcessFunction 底层API：注册/删除/使用 注册器
  * @Date: 2020/4/20 20:17 
  */
object ProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    //1.environment
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)   //设置时间语义类型

    //checkpoint--1:设置状态后端的 类型
    //    environment.setStateBackend(new FsStateBackend("file:///tmp/checkpoints"))
    //    environment.setStateBackend(new RocksDBStateBackend(""))

    //checkpoint--2: 检查点的运行配置
    //状态后端 设置 检查点 存储的时间间隔
    environment.enableCheckpointing(1000)
    //状态后端 设置 检查点 存储的时间间隔 ---和上面的一样
//    environment.getCheckpointConfig.setCheckpointInterval(1000)
    //设置检查点模式：状态一致性
    environment.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE)
    //设置超时时间，超时后丢弃当前checkpoint
    environment.getCheckpointConfig.setCheckpointTimeout(10000)
    //true: 若检查点操作异常，则当前整个JobManager会fail掉；false: 因为会存在保存checkpoint出异常，但是数据处理正常不影响正常运转 情况
    environment.getCheckpointConfig.setFailOnCheckpointingErrors(false)
    //设置最大同时运行几个checkpoint，默认是1
    environment.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    //两次checkpoint的最小时间间隔，与上面一条存在冲突：如果最大并行运行checkpoint设为1也设置两次最小时间间隔，则上面一条会失效
    environment.getCheckpointConfig.setMinPauseBetweenCheckpoints(100)
    //开启外部checkpoint持久化，一般是JobManager失败了会自动清理掉checkpoint,配置了此配置则不会清理掉 需手动清理
    //ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION：手动取消了JobManager则清除checkpoint
    //ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION：手动取消了JobManager，则仍然保留checkpoint
    environment.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION)

    //checkpoint--3: 配置重启策略
    //方式一：  固定延时重启处理:一共重启3次，每次间隔500毫秒
    environment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 500) )
    //方式二：失败率重启：300秒内重启3次，每次间隔10秒种
//    environment.setRestartStrategy(RestartStrategies.failureRateRestart(3, org.apache.flink.api.common.time.Time.seconds(300), org.apache.flink.api.common.time.Time.seconds(10)))





    // 2.source
    //    val inputStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")
    val inputStream = environment.socketTextStream("10.16.26.16",7777)

    // 3.transform 操作，但没有标记返回类型
    val dataStream = inputStream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
      //设置时间戳
      //处理乱序数据，延迟1秒钟上涨水位的操作(当时间到第11秒时，则认为第10秒为处理时间)
      //EventTime 的时钟设置 = (element.timestamp * 1000) - 1
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
      override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000
    })

    //底层API：ProcessFunction
    val processStream = dataStream.keyBy(_.id).process(new TemperatureIncreaseAlert())  //连续升温警报

    //状态后端 的操作 : 超过指定的温度差(升温或降温)会报警
    //方式一：由于processfunction，由于processfunction是个大招啥都能才，用一个小的模拟富含数
    val processStream2 = dataStream.keyBy(_.id).process(new MyTemperatureChangeAlert(10.0))
    //方式二：使用自定义RichFlatMapFunction
    val processStream3 = dataStream.keyBy(_.id).flatMap(new MyFlatMapTemperatureChangeAlert(10.0))
    //方式三：直接使用flatMapWithState
    val processStream4 = dataStream.keyBy(_.id).flatMapWithState[(String, Double, Double), Double] {
      //如果没有状态的话，也就是没有数据来过，那就将当前数据温度值存入状态
      case (input: SensorReading, None) => ( List.empty, Some(input.temperature) )
        //如果有状态，就应该与上次的温度比较差值，如果差值大于阈值就报警
      case ( input: SensorReading, lastTemp: Some[Double] ) =>
        val diff = (input.temperature - lastTemp.get).abs
        if (diff > 10){
          (List((input.id, lastTemp.get, input.temperature)), Some(input.temperature))
        }else
          ( List.empty, Some(input.temperature) )
    }
//    dataStream.print("input data")
//    processStream.print("process data")
//    processStream2.print("processStream2 data")
//    processStream3.print("processStream3 data")
    processStream4.print("processStream4 data")


    environment.execute("window test")

  }

}
//自定义ProcessFunction：连续升温警报
class TemperatureIncreaseAlert extends KeyedProcessFunction[String, SensorReading, String] {

  //定义一个状态，用来保存上一个数据的温度值
  lazy val lastTemperature: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemperature", classOf[Double]))

  //定义一个状态，用来存放定时器的时间戳
  lazy val currentTimer: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("currentTimer", classOf[Long]))

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {
    //先取出上一个温度值
    val preTemperature = lastTemperature.value()
    //更新温度值
    lastTemperature.update( value.temperature)

    //定时器的时间戳
    val curTimerTs = currentTimer.value()

    //温度上升 且 没有设过定时器，则注册定时器.  -->在注册定时器的10秒种内只存在温度上升且不存温度下降，则触发回调函数
    if (value.temperature > preTemperature && curTimerTs == 0){
      //定时器的时间戳： = 计时器的获取当前事件的处理时间：再加上10秒
      val timerTs = ctx.timerService().currentProcessingTime() + 10000L
      //注册processingTime事件
      ctx.timerService().registerProcessingTimeTimer( timerTs )
      //更新定时器的时间戳
      currentTimer.update(timerTs)

    }else if (value.temperature < preTemperature || preTemperature == 0.0){
      //如果温度下降，或是第一条数据，删除定时器并清空状态：根据注册定时器的时间戳去删除
      ctx.timerService().deleteProcessingTimeTimer( curTimerTs )
      //清空状态：不然一直增长会撑爆内存
      currentTimer.clear()
    }
  }

  //回调函数
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    //输出报警信息
    out.collect(ctx.getCurrentKey + "温度连续上升")
    currentTimer.clear()
  }
}

//自定义函数->采用ProcessFunction：超过指定温度阈值报警
class MyTemperatureChangeAlert(thresholdTemperature: Double) extends KeyedProcessFunction[String, SensorReading, (String, Double, Double)]{

  lazy val lastTemperature: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemperature",classOf[Double]))

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, (String, Double, Double)]#Context, out: Collector[(String, Double, Double)]): Unit = {
    //获取上一次的温度值
    val lastTemp = lastTemperature.value()
    //用当前的温度和上次的温度求差，如果大于阈值，输出报警信息
    val diff = (value.temperature - lastTemp).abs  //取绝对值

    if (diff >= thresholdTemperature){
      out.collect( (value.id, lastTemp, value.temperature) )
    }

    lastTemperature.update(value.temperature)
  }
}
//自定义函数->采用RichFlatMap：超过指定温度阈值报警
class MyFlatMapTemperatureChangeAlert(thresholdTemperature: Double) extends RichFlatMapFunction[SensorReading,(String, Double,Double)]{

  private var lastTemperature: ValueState[Double] = _

  override def open(parameters: Configuration): Unit = {
    //初始化的时候声明state变量
    lastTemperature = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemperature",classOf[Double]))
  }

  override def flatMap(value: SensorReading, out: Collector[(String, Double, Double)]): Unit = {
    //获取上一次的温度值
    val lastTemp = lastTemperature.value()
    //用当前的温度和上次的温度求差，如果大于阈值，输出报警信息
    val diff = (value.temperature - lastTemp).abs   //取绝对值

    if (diff >= thresholdTemperature){
      out.collect( (value.id, lastTemp, value.temperature) )
    }

    lastTemperature.update(value.temperature)
  }
}

